from kafka import KafkaProducer
import json
import time

def send_data(bootstrap_servers, topic_name, data):
	producer = KafkaProducer(bootstrap_servers=bootstrap_servers, value_serializer=lambda v: json.dumps(v).encode('utf-8'))
	future = producer.send(topic_name, data)
	try:
		record_metadata = future.get(timeout=10)

		print("future对象:",record_metadata)
		print("======================================================")
		print("接收的topic:",record_metadata.topic)
		print("partition_ID:",record_metadata.partition)
		print("offset:",record_metadata.offset)
	except Exception as e:
		print(str(e))
	print("==========发送成功============")
	producer.close()

def send_method_1000s(bootstrap_servers, topic_name):
	producer = KafkaProducer(bootstrap_servers=bootstrap_servers, value_serializer=lambda v: json.dumps(v).encode('utf-8'))

	for i in range(1000):
		data = {str(i): i}
		print(data)
		future = producer.send(topic_name, data)
		try:
			record_metadata = future.get(timeout=10)
			print("future对象:", record_metadata)
			print("======================================================")
			print("接收的topic:", record_metadata.topic)
			print("partition_ID:", record_metadata.partition)
			print("offset:", record_metadata.offset)
		except Exception as e:
			print(str(e))
		print("==========发送成功============")

		time.sleep(1)
	# 关闭生产者
	producer.close()






def send_method_bigjson(bootstrap_servers, topic_name):
	producer = KafkaProducer(bootstrap_servers=bootstrap_servers,
							 value_serializer=lambda v: json.dumps(v).encode('utf-8'))

	json_1 = json_1 = {
	"car_profile": {
		"license_plate": "浙FX632W",
		"time": {
			"timestamp": 1560764120,
			"details": {
				"image_url": "00002ce6-6d16-48c8-9030-96c6a130fd1d.jpg",
				"roadno": 12,
				"camerano": 300002,
				"laneno": 2,
				"orient": 1,
				"car_feature": "Zz/hR/NHLsxQzVlKuksZxSzISbVRy57MTslBTN7CXqzN0A7JE8dQTj1FVk4hSErIEc8+zIrRUUy5yK27oMzZSw==",
				"brand_id": 1,
				"model_id": "1442",
				"year_id": 698,
				"color_id": 7
			}
		}
	},
	"car_profile1": {
		"license_plate": "粤A2632W",
		"time": {
			"timestamp": 1560764120,
			"details": {
				"image_url": "00002ce6-6d16-48c8-9030-96c6a130fd1d.jpg",
				"roadno": 12,
				"camerano": 300002,
				"laneno": 2,
				"orient": 1,
				"car_feature": "Zz/hR/NHLsxQzVlKuksZxSzISbVRy57MTslBTN7CXqzN0A7JE8dQTj1FVk4hSErIEc8+zIrRUUy5yK27oMzZSw==",
				"brand_id": 1,
				"model_id": "1442",
				"year_id": 698,
				"color_id": 7
			}
		}
	},
	"car_profile2": {
		"license_plate": "浙FX632W",
		"time": {
			"timestamp": 1560764120,
			"details": {
				"image_url": "00002ce6-6d16-48c8-9030-96c6a130fd1d.jpg",
				"roadno": 12,
				"camerano": 300002,
				"laneno": 2,
				"orient": 1,
				"car_feature": "Zz/hR/NHLsxQzVlKuksZxSzISbVRy57MTslBTN7CXqzN0A7JE8dQTj1FVk4hSErIEc8+zIrRUUy5yK27oMzZSw==",
				"brand_id": 1,
				"model_id": "1442",
				"year_id": 698,
				"color_id": 7
			}
		}
	},
	"car_profile3": {
		"license_plate": "浙FX632W",
		"time": {
			"timestamp": 1560764120,
			"details": {
				"image_url": "00002ce6-6d16-48c8-9030-96c6a130fd1d.jpg",
				"roadno": 12,
				"camerano": 300002,
				"laneno": 2,
				"orient": 1,
				"car_feature": "Zz/hR/NHLsxQzVlKuksZxSzISbVRy57MTslBTN7CXqzN0A7JE8dQTj1FVk4hSErIEc8+zIrRUUy5yK27oMzZSw==",
				"brand_id": 1,
				"model_id": "1442",
				"year_id": 698,
				"color_id": 7
			}
		}
	},
	"car_profile4": {
		"license_plate": "粤A2632W",
		"time": {
			"timestamp": 1560764120,
			"details": {
				"image_url": "00002ce6-6d16-48c8-9030-96c6a130fd1d.jpg",
				"roadno": 12,
				"camerano": 300002,
				"laneno": 2,
				"orient": 1,
				"car_feature": "Zz/hR/NHLsxQzVlKuksZxSzISbVRy57MTslBTN7CXqzN0A7JE8dQTj1FVk4hSErIEc8+zIrRUUy5yK27oMzZSw==",
				"brand_id": 1,
				"model_id": "1442",
				"year_id": 698,
				"color_id": 7
			}
		}
	},
	"car_profile5": {
		"license_plate": "浙FX632W",
		"time": {
			"timestamp": 1560764120,
			"details": {
				"image_url": "00002ce6-6d16-48c8-9030-96c6a130fd1d.jpg",
				"roadno": 12,
				"camerano": 300002,
				"laneno": 2,
				"orient": 1,
				"car_feature": "Zz/hR/NHLsxQzVlKuksZxSzISbVRy57MTslBTN7CXqzN0A7JE8dQTj1FVk4hSErIEc8+zIrRUUy5yK27oMzZSw==",
				"brand_id": 1,
				"model_id": "1442",
				"year_id": 698,
				"color_id": 7
			}
		}
	}
	}




	for i in range(10000):
		t = time.strftime(r"%Y-%m-%d_%H:%M:%S", time.localtime())
		a = i
		t_dict = {"id": a, "time" : t}
		t_json = dict(t_dict,**json_1)

		try:
			producer.send(topic_name, t_json)

			# send_data(bootstrap_servers, topic_name, t_json)
			print(t, "id:%s" %a)
			print("="*50)
			# time.sleep(0.1)
		except Exception as e:
			print(e)
			producer.send(topic_name, t_json)
			print(t)
			print("=" * 50)
			# time.sleep(0.1)

	producer.close()

if __name__ == '__main__':

	conf = {
		'bootstrap_servers': '192.168.0.221:9092',
		'topic_name': 'test0210',
		'topic_name_a': 'test0210a'
	}

	bootstrap_servers = conf['bootstrap_servers']
	topic_name = conf['topic_name']
	data = "hello kafka!"

	# 发送一次一条数据
	# send_data(bootstrap_servers, topic_name, data)
	# 每秒发送一次数据
	# send_method_1000s(bootstrap_servers, topic_name)
	# 连续快速发送Json数据
	send_method_bigjson(bootstrap_servers, topic_name)










